package cn.jly.flink.connector;

import cn.jly.flink.constant.JdbcArgs;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

/**
 * jdbc sink
 *
 * @author lanyangji
 * @create 2020-09-03 20:25
 */
public class JdbcConnectorApp {
    /**
     * id生成器
     */
    private static final AtomicLong ID_GENERATOR = new AtomicLong(1000);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                new Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)),
                new Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)),
                new Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5)),
                new Person(ID_GENERATOR.incrementAndGet(), UUID.randomUUID().toString().substring(0, 5))
        )
                //.printToErr().setParallelism(1);
                .addSink(JdbcSink.sink(
                        "insert into person(name, age, email) values(?, ?, ?)",
                        (JdbcStatementBuilder<Person>) (preparedStatement, person) -> {
                            preparedStatement.setString(1, person.name);
                            preparedStatement.setInt(2, person.age);
                            preparedStatement.setString(3, person.email);
                        },
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl(JdbcArgs.url)
                                .withDriverName(JdbcArgs.driverName)
                                .withUsername(JdbcArgs.username)
                                .withPassword(JdbcArgs.password)
                                .build()
                ));

        env.execute("JdbcConnectorApp");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Person {
        private Long id;
        private String name;
        private Integer age;
        private String email;

        public Person(Long id, String name) {
            this.id = id;
            this.name = name;
            this.age = new Random().nextInt(40);
            this.email = this.name + "@qq.com";
        }
    }
}
